// Copyright 2013 Tumblr, Inc.
// Copyright 2013 Petar Maymounkov
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package hifi

import (
	x "circuit/sys/teleport"
	"math/rand"
	"sync"
)

// Within a physical connection, the connID distinguishes a unique logical session
type connID int64

func chooseConnID() connID {
	return connID(rand.Int63())
}

// conn is a read/write/closer for arbitrary Go values (i.e. interface{})
// conn is the user-facing front-end to a logical connection,
// maintained by the connection management system.
type conn struct {
	addr   x.Addr
	id     connID
	scrb   func()

	under__ sync.Mutex
	under   ReadWriteCloser

	wch__ sync.Mutex
	wch   chan interface{} // link.readLoop send msgs for this conn to ch
	rch   <-chan interface{}
}

type dialconn struct {
	conn
}

type acceptconn struct {
	conn
	sync.Mutex
	shook bool
}

const ConnReadBufferLen = 100

func dialConn(addr x.Addr, id connID, under ReadWriteCloser, scrb func()) *dialconn {
	ch := make(chan interface{}, ConnReadBufferLen)
	return &dialconn {
		conn: conn{
			addr:   addr,
			id:     id,
			scrb:   scrb,
			under:  under,
			wch:   ch,
			rch:   ch,
		},
	}
}

func acceptConn(addr x.Addr, id connID, under ReadWriteCloser, onclose func()) *acceptconn {
	ch := make(chan interface{}, ConnReadBufferLen)
	return &acceptconn{
		conn: conn{
			addr:  addr,
			id:    id,
			scrb:  onclose,
			under: under,
			wch:   ch,
			rch:   ch,
		},
		shook: false,
	}
}

func (c *conn) Read() (interface{}, error) {
	v, ok := <-c.rch
	if !ok {
		return nil, ErrClosed
	}
	return v, nil
}

func (c *acceptconn) handshake() error {
	c.Lock()
	defer c.Unlock()
	if c.shook {
		return nil
	}
	c.conn.under__.Lock()
	under := c.under
	c.conn.under__.Unlock()
	if under == nil {
		// Implies conn closed or in the process of closing
		return ErrClosed
	}
	c.shook = true
	return under.Write(&linkOpenMsg{ID: c.id})
}

func (c *acceptconn) Read() (interface{}, error) {
	if err := c.handshake(); err != nil {
		return nil, err
	}
	return c.conn.Read()
}

func (c *conn) Write(payload interface{}) error {
	c.under__.Lock()
	under := c.under
	c.under__.Unlock()
	if under == nil {
		return ErrClosed
	}
	msg := &linkConnMsg{ID: c.id, Payload: payload}
	return under.Write(msg)
}

func (c *acceptconn) Write(payload interface{}) error {
	if err := c.handshake(); err != nil {
		return err
	}
	return c.conn.Write(payload)
}

// The link object calls sendRead every time a new message is demultiplexed for this connection
func (c *conn) sendRead(v interface{}) error {
	// Ensure that only one instance of sendRead is pending before the under__ at a time
	c.wch__.Lock()
	defer c.wch__.Unlock()
	if c.wch == nil {
		// Implies conn closed or in the process of closing
		return ErrClosed
	}
	c.wch <- v
	return nil
}

func (c *conn) Close() error {
	// Close sendRead channel
	c.wch__.Lock()
	if c.wch == nil {
		c.wch__.Unlock()
		return ErrClosed
	}
	wch := c.wch
	c.wch = nil
	c.wch__.Unlock()

	// Drain the read channel before closing it, in order to unblock any outstanding sendRead (there can be at most one)
	select {
	case <-c.rch:
	default:
	}
	close(wch)

	// Close the underlying connection
	c.under__.Lock()
	c.under = nil
	c.under__.Unlock()

	// Call on-close hook once, outside of any locks.
	// There is no need for locking, since only one instance of Close is guaranteed to reach here
	if c.scrb != nil {
		c.scrb()
	}
	c.scrb = nil

	return nil
}

func (c *conn) RemoteAddr() x.Addr {
	return c.addr
}
